#include "StdAfx.h"
#include "RstreamOperator.h"
#include "QueueEntity.h"
#include "DocumentIdentifierGenerator.h"

RstreamOperator::RstreamOperator(void)
{
	this->relationSynopsis.reset(new RelationSynopsis());
	this->lastInputTimestamp = 0;
	this->elementIterator.reset(new ElementIterator(this->relationSynopsis));
}


RstreamOperator::~RstreamOperator(void)
{
}


void RstreamOperator::execution(int executionTimes)
{
#ifdef DEBUG
	std::cout<<"===================operator begin================="<<std::endl;
	std::cout<<"operatorid : "<<this->getId()<<std::endl;
#endif
	assert(this->getInputQueueList().size()==1);
	assert(this->getOutputQueueList().size()==1);
	boost::shared_ptr<QueueEntity>inputQueue = this->getInputQueueList().front();
	boost::shared_ptr<QueueEntity>outputQueue = this->getOutputQueueList().front();
	if(this->getOperatorState()==BLOCKED)
	{
		dealWithBlock();
		if(this->getOperatorState()==BLOCKED)
		{
#ifdef DEBUG
			std::cout<<"operator blocked"<<std::endl;
			std::cout<<"===================operator over================="<<std::endl;
#endif
			return;
		}
	}
	Element inputElement;
	for(int i = 0 ; i<executionTimes ; i++)
	{
		if(outputQueue->isFull())
		{
			break;
		}
		if(inputQueue->isEmpty())
		{
			break;
		}
		inputQueue->dequeue(inputElement);
		for(;lastInputTimestamp<inputElement.timestamp;lastInputTimestamp++)
		{
			produceOutput();
			if(this->getOperatorState()==BLOCKED)
			{
				this->blockElement = inputElement;
#ifdef DEBUG
				std::cout<<"operator blocked"<<std::endl;
				std::cout<<"===================operator over================="<<std::endl;
#endif
				return;
			}
		}


		if(inputElement.mark==PLUS_MARK)
		{
			this->relationSynopsis->insertElement(inputElement);
		}
		else if(inputElement.mark==MINUS_MARK)
		{
			this->relationSynopsis->deleteElement(inputElement);
		}


	}


#ifdef DEBUG
	std::cout<<"===================operator over================="<<std::endl;
#endif
}
void RstreamOperator::produceOutput()
{

	boost::shared_ptr<QueueEntity>outputQueue = this->getOutputQueueList().front();

	this->elementIterator->initial();
	Element elementInSynopsis;
	while((!outputQueue->isFull())&&this->elementIterator->getNext(elementInSynopsis))
	{
		Element outputElement;
		outputElement.document = elementInSynopsis.document;
		outputElement.id = elementInSynopsis.id;
		outputElement.mark = PLUS_MARK;
		outputElement.timestamp = lastInputTimestamp;
		outputQueue->enqueue(outputElement);
	}
	if(outputQueue->isFull())
	{
		assert(this->getOperatorState()==NORMAL);
		this->setOperatorState(BLOCKED);
		
	}
}

void RstreamOperator::dealWithBlock(void)
{
	
	boost::shared_ptr<QueueEntity>outputQueue = this->getOutputQueueList().front();
	Element elementInSynopsis;
	while((!outputQueue->isFull())&&this->elementIterator->getNext(elementInSynopsis))
	{
		Element outputElement;
		outputElement.document = elementInSynopsis.document;
		outputElement.id = elementInSynopsis.id;
		outputElement.mark = PLUS_MARK;
		outputElement.timestamp = lastInputTimestamp;
		outputQueue->enqueue(outputElement);
	}
	if(outputQueue->isFull())
	{
		return;
	}
	this->setOperatorState(NORMAL);
	lastInputTimestamp++;
	assert(lastInputTimestamp<this->blockElement.timestamp);
	for(;lastInputTimestamp<this->blockElement.timestamp;lastInputTimestamp++)
	{
		produceOutput();
		if(this->getOperatorState()==BLOCKED)
		{
			return;
		}

	}
	if(blockElement.mark==PLUS_MARK)
	{
		this->relationSynopsis->insertElement(blockElement);
	}
	else if(blockElement.mark==MINUS_MARK)
	{
		this->relationSynopsis->deleteElement(blockElement);
	}
}